#!/bin/bash

function task_job() {
jdbcUrl="$1"
db="$2"
mysqlTable="$3"
hiveTable="$4"
column="${5}"
sub="${6}"
threadNumber="${7}"
typed="${8}"
dt="{{ execution_date | cst_ds }}"
nextdt="{{ execution_date | date_add(1) | cst_ds }}"
#dt="{{ execution_date | date_add(-5) | cst_ds }}"
#nextdt="{{ execution_date | date_add(-4) | cst_ds }}"
env="{{ var.value.env_sync }}"
username='{{ var.json.mysql_spmi.username }}'
password='{{ var.json.mysql_spmi.password }}'

json='{
"reader":{
"connect":{
"url":"'"${jdbcUrl}"'",
"username":"'"${username}"'",
"password":"'"${password}"'",
"driver":"com.mysql.cj.jdbc.Driver"
},
"dbtype":"mysql",
"tableName":"'"${db}.${mysqlTable}"'",
"subTableList":"'"${sub}"'",
"where":"'"${column} between str_to_date('${dt} 00:00:00', '%Y-%m-%d %H:%i:%s') and str_to_date('${nextdt} 00:00:00', '%Y-%m-%d %H:%i:%s')"'",
"query":"'""'",
"splitColumn":"'"${column}"'",
"equalitySectioning":0,
"containsnull":0,
"fetchsize":"1024",
"threadNumber":0
},
"channel":{
"filterAbnormalCharacter":0
},
"writer":{
"dag":"spmi_ods",
"task_id":"spmi_ods__spmi_transfer_bill",
"dbtype":"hive",
"tableName":"'"${hiveTable}"'",
"database":"spmi_ods",
"writeMode": "'"${typed}"'",
"partitionColumn":"dt",
"partitionValue":"'"${dt}"'"},
"settting":{
"env":"'"${env}"'"}
}'

/usr/hdp/3.1.4.0-315/spark2/bin/spark-submit \
--master yarn \
--deploy-mode cluster \
--driver-memory 18G \
--driver-cores 4 \
--executor-memory 5G \
--executor-cores 1 --queue pro  \
--num-executors ${threadNumber}  \
--conf spark.driver.maxResultSize=6g \
--class com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver \
hdfs:///scheduler/jms/spark/sync/mysql/spark_sync.jar  "$json"
echo ${json} >> /jms_ods/spmi_transfer_bill.log
echo "任务执行成功: ${dt} ${hiveTable} ${db} ${sub}"
}

jdbcUrl1='{{ var.json.mysql_spmi.db03 }}'
jdbcUrl2='{{ var.json.mysql_spmi.db_tof1 }}'
jdbcUrl3='{{ var.json.mysql_spmi.db_tof2 }}'
jdbcUrl4='{{ var.json.mysql_spmi.db_tof3 }}'
mTab="spmi_transfer_bill"
hTab="spmi_transfer_bill"
col='update_time'

# 数据已清空2023-02-03
#task_job  ${jdbcUrl1}  yl_spmibill_8_1   ${mTab} ${hTab} ${col} "[1,2];[0,7]"  8  overwrite
#task_job  ${jdbcUrl1}  yl_spmibill_8_1   ${mTab} ${hTab} ${col} "[3,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_2   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_3   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_4   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_5   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_6   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_7   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_8   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_9   ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_10  ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_11  ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &
#task_job  ${jdbcUrl1}  yl_spmibill_8_12  ${mTab} ${hTab} ${col} "[1,31];[0,7]" 4  append    &

wait

#2022-01-20号新加库表实例
task_job  ${jdbcUrl1}  yl_spmibill_158  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

task_job  ${jdbcUrl1}  yl_spmibill_168  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

task_job  ${jdbcUrl2}  yl_spmibill_158  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl2}  yl_spmibill_168  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

task_job  ${jdbcUrl3}  yl_spmibill_158  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl3}  yl_spmibill_168  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

task_job  ${jdbcUrl4}  yl_spmibill_158  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl4}  yl_spmibill_168  ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

# 2022-09-06 新增分库和表
task_job  ${jdbcUrl1}  yl_spmibill_tof_1   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl1}  yl_spmibill_tof_2   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl1}  yl_spmibill_tof_3   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

task_job  ${jdbcUrl2}  yl_spmibill_tof_1   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl2}  yl_spmibill_tof_2   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl2}  yl_spmibill_tof_3   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

task_job  ${jdbcUrl3}  yl_spmibill_tof_1   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl3}  yl_spmibill_tof_2   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl3}  yl_spmibill_tof_3   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

task_job  ${jdbcUrl4}  yl_spmibill_tof_1   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl4}  yl_spmibill_tof_2   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &
task_job  ${jdbcUrl4}  yl_spmibill_tof_3   ${mTab} ${hTab} ${col} "[0,255]" 4  append    &

wait

echo "${hTab} 表所有任务执行成功! "
